package com.openx.amqp;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor;
import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Role;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Configuration
@EnableConfigurationProperties({XRabbitProperties.class, XRabbitMutiProperties.class})
public class XRabbitAutoConfiguration {
    public XRabbitAutoConfiguration(){
        System.out.println("XRabbitAutoConfiguration init");
    }
    @Autowired
    private XRabbitMutiProperties xRabbitMutiProperties;
    
    @Autowired
    private XRabbitProperties xRabbitProperties;
    
    @Autowired
    private ApplicationContext ac;
    
    @Resource
    private SimpleRabbitListenerContainerFactoryConfigurer configurer;
    
    @Bean
    public ConnectionFactory connectionFactory() throws Exception {
        System.out.println("[CQY]加载connectionFactory");
        CachingConnectionFactory connectionFactory = null;
        //获取BeanFactory
        DefaultListableBeanFactory dbf = (DefaultListableBeanFactory) ac.getAutowireCapableBeanFactory();
        if (xRabbitMutiProperties.isEnable()) {
            // 多rabbitmq
            XRabbitProperties[] rps = xRabbitMutiProperties.getRabbit();
            if (rps != null && rps.length > 0) {
                // 名称
                String factoryName = null, tempName = null, messageListenerName = null, adminName = null, listenerFactoryName = null;
                BeanDefinitionBuilder beanDef = null;
                for (int i = rps.length - 1; i > -1; i--) {
                    XRabbitProperties rp = rps[i];
                    //创建bean信息.
                    beanDef =
                        BeanDefinitionBuilder.genericBeanDefinition(CachingConnectionFactory.class);
                    beanDef.addPropertyValue("host", rp.getHost());
                    beanDef.addPropertyValue("port", rp.getPort());
                    beanDef.addPropertyValue("username", rp.getUsername());
                    beanDef.addPropertyValue("password", rp.getPassword());
                    beanDef.addPropertyValue("virtualHost", rp.getVirtualHost());
                    beanDef.addPropertyValue("publisherConfirms", rp.getPublisherConfirms()); //必须要设置
                    
                    //动态注册bean.
                    if (rp.getName() == null || "".equals(rp.getName().trim())) {
                        factoryName = "connectionFactory" + i;
                        tempName = "rabbitTemplate" + i;
                        messageListenerName = "replyMessageListener" + i;
                        adminName = "rabbitAdmin" + i;
                        listenerFactoryName = "containerFactory" + i;
                    } else {
                        factoryName = rp.getName() + "ConnectionFactory";
                        tempName = rp.getName() + "RabbitTemplate";
                        messageListenerName = rp.getName() + "ReplyMessageListener";
                        adminName = rp.getName() + "RabbitAdmin";
                        listenerFactoryName = rp.getName() + "ContainerFactory";
                    }
                    System.out.println("[CQY] 注入connectionFactory-" + factoryName);
                    dbf.registerBeanDefinition(factoryName, beanDef.getBeanDefinition());
                    connectionFactory = dbf.getBean(factoryName, CachingConnectionFactory.class);
                    // 声明队列、交换、路由
                    rabbitAdmin(dbf, connectionFactory, rp, adminName);

                    // 创建RabbitTemplate
                    injectRabbitTemplate(0 == i, dbf, connectionFactory, tempName);

                    // listener
                    containerFactory(dbf, connectionFactory, listenerFactoryName);

                    // 创建监听RCP模式
                    rabbitTemplateListeners(dbf, rp, connectionFactory, tempName, messageListenerName);
                }
            }
        } else {
            // 单rabbitmq
            connectionFactory = new CachingConnectionFactory();
            connectionFactory.setHost(xRabbitProperties.getHost());
            connectionFactory.setPort(xRabbitProperties.getPort());
            connectionFactory.setUsername(xRabbitProperties.getUsername());
            connectionFactory.setPassword(xRabbitProperties.getPassword());
            connectionFactory.setVirtualHost(xRabbitProperties.getVirtualHost());
            connectionFactory.setPublisherConfirms(xRabbitProperties.getPublisherConfirms()); //必须要设置
            // 声明队列、交换、路由
            rabbitAdmin(dbf, connectionFactory, xRabbitProperties, null);
            
            // 创建RabbitTemplate
            injectRabbitTemplate(true, dbf, connectionFactory,null);

            // listener
            containerFactory(dbf, connectionFactory, null);

            // 创建监听RCP模式
            rabbitTemplateListeners(dbf, xRabbitProperties, connectionFactory, null, null);
        }
        

        return connectionFactory;
    }

    public void rabbitTemplateListeners(DefaultListableBeanFactory dbf, XRabbitProperties rp, CachingConnectionFactory connectionFactory, String templateName, String messageListenerName) {
        if(rp.getRpcListenerQueues() == null || rp.getRpcListenerQueues().trim().equals("")) {
            return;
        }
        if(templateName == null) {
            templateName = "rabbitTemplate";
        }
        if(messageListenerName == null) {
            messageListenerName = "replyMessageListener";
        }
        RabbitTemplate rabbitTemplate = dbf.getBean(templateName, RabbitTemplate.class);
        List<String> rpcQueuesList = new ArrayList<String>();
        try {
            String[] rpcQueues = rp.getRpcListenerQueues().trim().split(":");
            String exchangeName = rpcQueues[0];
            rpcQueues = rpcQueues[1].trim().split(",");
            for (String rpcQueue : rpcQueues) {
                if (rpcQueue == null || rpcQueue.trim().equals("")) {
                    continue;
                }
                rpcQueuesList.add(rpcQueue.trim());
            }
            System.out.println("[CQY] 开启RPC模式-" + templateName);

            // 配置监听exchange
            rabbitTemplate.setExchange(exchangeName);
            // 这里必须这么设置，如果设置成队列名就只能监听1个队列。
            rabbitTemplate.setReplyAddress(exchangeName + "/*");
        } catch (Exception e) {
            throw new RuntimeException("rpc-listener-queues配置不正确。例如 exchange:queue1,queue2");
        }
        if(rpcQueuesList.size() == 0) {
            return;
        }

        System.out.println("[CQY] 注入messageListenerContainer-" + messageListenerName);
        // 创建SimpleMessageListenerContainer
        BeanDefinitionBuilder beanDef = BeanDefinitionBuilder.genericBeanDefinition(SimpleMessageListenerContainer.class);
        beanDef.addConstructorArgValue(connectionFactory);
        beanDef.addPropertyValue("messageListener", rabbitTemplate);

        Queue[] rpcQueuesArr = new Queue[rpcQueuesList.size()];
        for (int i = 0; i < rpcQueuesList.size() ; i++) {
            rpcQueuesArr[i] = new Queue(rpcQueuesList.get(i).trim());
        }
        beanDef.addPropertyValue("queues", rpcQueuesArr);

        dbf.registerBeanDefinition(messageListenerName, beanDef.getBeanDefinition());
    }

    // 参数ConnectionFactory connectionFactory，目的是要在扫描@RabbitListener之前注入。
    @Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public RabbitListenerAnnotationBeanPostProcessor rabbitListenerAnnotationProcessor(ConnectionFactory connectionFactory) throws Exception {
        System.out.println("rabbit1");
        return new RabbitListenerAnnotationBeanPostProcessor();
    }

    @Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
    public RabbitListenerEndpointRegistry defaultRabbitListenerEndpointRegistry() {
        System.out.println("rabbit2");
        return new RabbitListenerEndpointRegistry();
    }
    
    private void injectRabbitTemplate(boolean isPrimary, DefaultListableBeanFactory dbf, CachingConnectionFactory connectionFactory, String tempName){
        // 创建RabbitTemplete
        BeanDefinitionBuilder beanDef = BeanDefinitionBuilder.genericBeanDefinition(RabbitTemplate.class);
        beanDef.addConstructorArgValue(connectionFactory);
        if(tempName == null) {
            tempName = "rabbitTemplate";
        }
        System.out.println("[CQY] 注入rabbitTemplate-" + tempName);
        beanDef.setPrimary(isPrimary);
        if(isPrimary) {
            System.out.println("[CQY] 设置为Primary-" + tempName);
        }
        dbf.registerBeanDefinition(tempName, beanDef.getBeanDefinition());
    }
    
    /**
     * 生成Queue
     * @author quanyou.chen
     * @date: 2017年11月1日 上午10:34:07
     * @param queues
     * @param queueMap
     * @return
     */
    private Map<String, Queue> createQueues(String queues, Map<String, Queue> queueMap) {
        if (queues != null && !"".equals(queues.trim())) {
            Queue queue = null;
            for (String queueName : queues.trim().split(",")) {
                if(queueMap.get(queueName) != null) {
                    continue;
                }
                queue = new Queue(queueName);
                queueMap.put(queueName, queue);
            }
        }
        return queueMap;
    }
    
    /**
     * 生成Exchanges
     * @author quanyou.chen
     * @date: 2017年11月1日 上午10:34:22
     * @param exchanges
     * @param exchangesMap
     * @return
     * @throws Exception
     */
    private Map<String, Exchange> createExchanges(String exchanges, Map<String, Exchange> exchangesMap) throws Exception {
        if (exchanges != null && !"".equals(exchanges.trim())) {
            String[] exchangeSp = null;
            Exchange exchange = null;
            XExchangeType type = null;
            for (String exchangeStr : exchanges.trim().split(",")) {
                exchangeSp = exchangeStr.split(":");
                if (exchangeSp.length == 1) {
                    type = XExchangeType.topic;
                } else {
                    type = XExchangeType.fromString(exchangeSp[1].toLowerCase());
                }
                if (type == null) {
                    throw new Exception("【" + exchangeSp[0] + "】交换机的类型【" + exchangeSp[1].toLowerCase() + "】不存在。");
                }
                if(exchangesMap.get(exchangeSp[0]) != null) {
                    continue;
                }
                exchange = new CustomExchange(exchangeSp[0], type.toString());
                exchangesMap.put(exchangeSp[0], exchange);
            }
        }
        return exchangesMap;
    }
    
    /**
     * 绑定交换队列路由
     * @author quanyou.chen
     * @date: 2017年11月1日 上午10:34:32
     * @param rabbitAdmin
     * @param exchange
     * @param routingKey
     * @param queue
     */
    private void declareBinding(RabbitAdmin rabbitAdmin, Exchange exchange, String routingKey, Queue queue) {
        //        system("system"),direct("direct"),fanout("fanout"),header("headers"),topic("topic");
        System.out.println("[CQY] 声明绑定关系-[" + exchange.getType() + ":" + exchange.getName() + "->" + routingKey + "->" + queue.getName() + "]");
        switch (XExchangeType.fromString(exchange.getType())) {
            case direct:
                DirectExchange direct = new DirectExchange(exchange.getName());
                rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(direct).with(routingKey));
                break;
            case topic:
                TopicExchange topic = new TopicExchange(exchange.getName());
                rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(topic).with(routingKey));
                break;
            case fanout:
                FanoutExchange fanout = new FanoutExchange(exchange.getName());
                rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(fanout));
                break;
            case header:
                //                HeadersExchange headers = new HeadersExchange(exchange.getName());
                //                rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(headers).);
                break;
            
            default:
                break;
        }
    }
    
    /**
     * 检查配置项是否正确
     * @author quanyou.chen
     * @date: 2017年11月1日 上午10:34:53
     * @param exchanges
     * @param routes
     * @param queues
     * @param exchangesMap
     * @param routesList
     * @param queueMap
     * @throws Exception
     */
    private void checkMQConfig(String exchanges, String routes, String queues, Map<String, Exchange> exchangesMap, List<XRoute> routesList, Map<String, Queue> queueMap) throws Exception{
        System.out.println("[CQY] 校验MQ配置");
        // Queue
        createQueues(queues, queueMap);
        // Exchange
        createExchanges(exchanges, exchangesMap);
        
        // Route
        if (routes != null && !"".equals(routes.trim())) {
            String[] routeSp = null;
            Exchange exchange = null;
            Queue queue = null;
            XRoute route = null;
            for (String routeStr : routes.split(",")) {
                routeSp = routeStr.split(":");
                String exchangeName = routeSp[0];
                String routeKey = routeSp[1];
                String queueName = null;
                if(routeSp.length == 3) {
                    queueName = routeSp[2];
                }
                
                if (exchangesMap != null && exchangesMap.size() > 0) {
                    exchange = exchangesMap.get(exchangeName);
                    if (exchange == null) {
                        // 异常
                        throw new Exception("路由配置“" + routeStr + "”中的“" + exchangeName + "”在【exchanges配置】中不存在。");
                    }
                } else {
                    throw new Exception("缺少配置【exchanges配置】。");
                }

                if(XExchangeType.fanout == XExchangeType.fromString(exchange.getType()) && queueName == null) {
                    // 广播，不需要routingKey，只有exchange和queue
                    queueName = routeKey;
                    routeKey = null;
                }

                if (queueMap != null && queueMap.size() > 0) {
                    queue = queueMap.get(queueName);
                    if (queue == null) {
                        // 异常
                        throw new Exception("路由配置“" + routeStr + "”中的“" + queueName + "”在【queues/rpc-listener-queues配置】中不存在。");
                    }
                } else {
                    queue = new Queue(queueName);
                }
                
                route = new XRoute(exchange, routeKey, queue);
                routesList.add(route);
            }
        }
    }
    
    /**
     * 声明队列、交换、路由
     * @author quanyou.chen
     * @date: 2017年11月1日 上午10:35:10
     * @param rabbitAdmin
     * @param exchangesMap
     * @param routesList
     * @param queueMap
     */
    private void declareMQConfig(RabbitAdmin rabbitAdmin, Map<String, Exchange> exchangesMap, List<XRoute> routesList, Map<String, Queue> queueMap) {
        // declare exchange
        for (Exchange exchange : exchangesMap.values()) {
            System.out.println("[CQY] 声明交换-" + exchange.getName());
            rabbitAdmin.declareExchange(exchange);
        }
        // declare queue
        for (Queue queue : queueMap.values()) {
            System.out.println("[CQY] 声明队列-" + queue.getName());
            rabbitAdmin.declareQueue(queue);
        }
        // declare route
        for (XRoute route : routesList) {
            declareBinding(rabbitAdmin, route.getExchange(), route.getRouteKey(), route.getQueue());
        }
    }
    
    /** 
     * rabbitAdmin代理类 
     * @return 
     * @throws Exception 
     */
    private RabbitAdmin rabbitAdmin(DefaultListableBeanFactory dbf, ConnectionFactory connectionFactory, XRabbitProperties rp, String adminName) throws Exception {
        RabbitAdmin rabbitAdmin;

        if (adminName == null) {
            adminName = "rabbitAdmin";
        }
        System.out.println("[CQY] 注入admin["+adminName+"]");
        if(rp.isDeclare()) {
            // 注册
            BeanDefinitionBuilder beanDef = BeanDefinitionBuilder.genericBeanDefinition(RabbitAdmin.class);
            beanDef.addConstructorArgValue(connectionFactory);
            beanDef.addPropertyValue("autoStartup", true);
            dbf.registerBeanDefinition(adminName, beanDef.getBeanDefinition());
            // 获取实例
            rabbitAdmin = dbf.getBean(adminName, RabbitAdmin.class);
        } else {
            rabbitAdmin = new RabbitAdmin(connectionFactory);
        }

        String exchanges = rp.getExchanges();
        String routes = rp.getRoutes();
        String queues = rp.getQueues();
        String listenerQueues = rp.getRpcListenerQueues();
        if(listenerQueues != null && !"".equals(listenerQueues.trim())) {
            queues += "," + listenerQueues.split(":")[1];
        }

        Map<String, Queue> queueMap = new HashMap<String, Queue>();
        Map<String, Exchange> exchangesMap = new HashMap<String, Exchange>();
        List<XRoute> routesList = new ArrayList<XRoute>();
        // check
        checkMQConfig(exchanges, routes, queues, exchangesMap, routesList, queueMap);
        if(rp.isDeclare()) {
            System.out.println("[CQY] 声明MQ队列和交换");
            // declare
            declareMQConfig(rabbitAdmin, exchangesMap, routesList, queueMap);
        }

        return rabbitAdmin;
    }
    
    /**
     * 生成listenerContainerFactory
     * @author quanyou.chen
     * @date: 2017年11月1日 上午11:18:16
     * @param connectionFactory
     * @return
     */
    public SimpleRabbitListenerContainerFactory containerFactory(DefaultListableBeanFactory dbf, ConnectionFactory connectionFactory, String listenerFactoryName) {
        if(listenerFactoryName == null) {
            listenerFactoryName = "containerFactory";
        }
        System.out.println("[CQY] 注入containerFactory-" + listenerFactoryName);
        //获取BeanFactory
        BeanDefinitionBuilder beanDef =
            BeanDefinitionBuilder.genericBeanDefinition(SimpleRabbitListenerContainerFactory.class);
        dbf.registerBeanDefinition(listenerFactoryName, beanDef.getBeanDefinition());
        SimpleRabbitListenerContainerFactory listenerFactory = dbf.getBean(listenerFactoryName, SimpleRabbitListenerContainerFactory.class);
        configurer.configure(listenerFactory, connectionFactory);
        return listenerFactory;
    }
}
